package store

import (
	"bytes"
	"encoding/binary"
	"errors"
	"github.com/boltdb/bolt"
	"learning/gooop/etcd/raft/model"
)

type tBoltDBStore struct {
	file               string
	lastAppendedTerm   int64
	lastAppendedIndex  int64
	lastCommittedTerm  int64
	lastCommittedIndex int64

	db *bolt.DB
}

func NewBoltStore(file string) (error, ILogStore) {
	db, err := bolt.Open(file, 0600, nil)
	if err != nil {
		return err, nil
	}

	store := new(tBoltDBStore)
	err = db.Update(func(tx *bolt.Tx) error {
		b, e := tx.CreateBucketIfNotExists(gMetaBucket)
		if e != nil {
			return e
		}

		v := b.Get(gKeyCommittedTerm)
		if v == nil {
			e = b.Put(gKeyCommittedTerm, int64ToBytes(gDefaultTerm))
			if e != nil {
				return e
			}
			store.lastCommittedTerm = gDefaultTerm

		} else {
			store.lastCommittedTerm = bytesToInt64(v)
		}

		v = b.Get(gKeyCommittedIndex)
		if v == nil {
			e = b.Put(gKeyCommittedIndex, int64ToBytes(gDefaultIndex))
			if e != nil {
				return e
			}
			store.lastCommittedIndex = gDefaultIndex

		} else {
			store.lastCommittedIndex = bytesToInt64(v)
		}

		b, e = tx.CreateBucketIfNotExists(gDataBucket)
		if e != nil {
			return e
		}

		b = tx.Bucket(gUnstableBucket)
		if b != nil {
			e = tx.DeleteBucket(gUnstableBucket)
			if e != nil {
				return e
			}
		}
		_, e = tx.CreateBucketIfNotExists(gUnstableBucket)
		if e != nil {
			return e
		}

		_, e = tx.CreateBucketIfNotExists(gCommittedBucket)
		if e != nil {
			return e
		}

		return nil
	})

	if err != nil {
		return err, nil
	}

	store.db = db
	return nil, store
}

func int64ToBytes(i int64) []byte {
	buf := bytes.NewBuffer(make([]byte, 8))
	_ = binary.Write(buf, binary.BigEndian, i)
	return buf.Bytes()
}

func bytesToInt64(data []byte) int64 {
	var i int64
	buf := bytes.NewBuffer(data)
	_ = binary.Read(buf, binary.BigEndian, &i)
	return i
}

func (me *tBoltDBStore) LastCommittedTerm() int64 {
	return me.lastCommittedTerm
}

func (me *tBoltDBStore) LastCommittedIndex() int64 {
	return me.lastCommittedIndex
}

func (me *tBoltDBStore) LastAppendedTerm() int64 {
	return me.lastAppendedTerm
}

func (me *tBoltDBStore) LastAppendedIndex() int64 {
	return me.lastAppendedIndex
}

func (me *tBoltDBStore) Append(entry *model.LogEntry) error {
	cmd := gCmdFactory.OfTag(entry.Tag)
	cmd.Unmarshal(entry.Command)

	e, entryData := entry.Marshal()
	if e != nil {
		return e
	}

	return me.db.Update(func(tx *bolt.Tx) error {
		// save log to unstable
		b := tx.Bucket(gUnstableBucket)
		e = b.Put(int64ToBytes(entry.Index), entryData)
		if e != nil {
			return e
		}

		return nil
	})
}

func (me *tBoltDBStore) Commit(index int64) error {
	return me.db.Update(func(tx *bolt.Tx) error {
		// read unstable log
		ub := tx.Bucket(gUnstableBucket)
		k := int64ToBytes(index)
		data := ub.Get(k)
		if data == nil {
			return gErrorCommitLogNotFound
		}

		entry := new(model.LogEntry)
		e := entry.Unmarshal(data)
		if e != nil {
			return e
		}

		// apply cmd
		cmd := gCmdFactory.OfTag(entry.Tag)
		cmd.Unmarshal(entry.Command)
		e = cmd.Apply(tx)
		if e != nil {
			return e
		}

		// save to committed log
		cb := tx.Bucket(gCommittedBucket)
		e = cb.Put(k, data)
		if e != nil {
			return e
		}

		// update committed.index, committed.term
		mb := tx.Bucket(gMetaBucket)
		e = mb.Put(gKeyCommittedIndex, int64ToBytes(index))
		if e != nil {
			return e
		}

		e = mb.Put(gKeyCommittedTerm, int64ToBytes(entry.Term))
		if e != nil {
			return e
		}

		// del unstable.index
		e = ub.Delete(k)
		if e != nil {
			return e
		}

		me.lastCommittedIndex = entry.Index
		me.lastCommittedTerm = entry.Term
		return nil
	})
}

func (me *tBoltDBStore) GetLog(index int64) (error, *model.LogEntry) {
	ret := []*model.LogEntry{nil}
	e := me.db.View(func(tx *bolt.Tx) error {
		k := int64ToBytes(index)
		v := tx.Bucket(gCommittedBucket).Get(k)

		if v == nil {
			return nil
		}

		entry := new(model.LogEntry)
		e := entry.Unmarshal(v)
		if e != nil {
			return e
		}

		ret[0] = entry
		return nil
	})

	return e, ret[0]
}

func (me *tBoltDBStore) Get(key []byte) (error, []byte) {
	ret := [][]byte{nil}
	e := me.db.View(func(tx *bolt.Tx) error {
		b := tx.Bucket(gDataBucket)
		ret[0] = b.Get(key)
		return nil
	})

	if e != nil {
		return e, nil
	} else {
		return nil, ret[0]
	}
}

func (me *tBoltDBStore) Put(key []byte, content []byte) error {
	return me.db.Update(func(tx *bolt.Tx) error {
		b := tx.Bucket(gDataBucket)
		return b.Put(key, content)
	})
}

func (me *tBoltDBStore) Del(key []byte) error {
	return me.db.Update(func(tx *bolt.Tx) error {
		b := tx.Bucket(gDataBucket)
		return b.Delete(key)
	})
}

var gMetaBucket = []byte("meta")
var gUnstableBucket = []byte("unstable")
var gCommittedBucket = []byte("committed")
var gDataBucket = []byte("data")

var gKeyCommittedIndex = []byte("committed.index")
var gKeyCommittedTerm = []byte("committed.term")

var gDefaultTerm int64 = 0
var gDefaultIndex int64 = 0

var gErrorCommitLogNotFound = errors.New("committing log not found")
